package com.ra.util;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.stereotype.Service;

import com.espirit.eap.manager.SystemConfig;
import com.espirit.eap.manager.SystemConfigService;
import com.espirit.eap.manager.TaskScheduler;
import com.espirit.eap.manager.TaskSchedulerService;
import com.googlecode.cswish.exception.ApplicationException;
import com.googlecode.cswish.struts.ContextHolder;
import com.googlecode.cswish.struts.spring.Executor;
import com.googlecode.cswish.struts.spring.GenericJPAService;
import com.googlecode.cswish.struts.spring.SystemEnv;

@Service
public class SyncSocketService {
	
	private static final Log logger = LogFactory.getLog(SyncSocketService.class);

	@Resource
	private SystemConfigService systemConfigService;
	
	@Resource
	private GenericJPAService genericJPAService;
	
	@Resource
	private TaskSchedulerService taskSchedulerService;
	
	private String ip = "localhost";// 服务器地址
	private int port = 5000;// 服务器端口号
	private Socket client;
	private Thread clientThread;
	
	private final static byte[] HEARTBEAT_COMMAND = new byte[] {0x01, 0x00, 0x00, 0x00};
	
	private TaskScheduler heartbeatScheduler;
	private TaskScheduler connectionScheduler;
	
	private Executor<Boolean, Object[]> commandCallback;
	
	private int errorCount;
	private List<byte[]> commandWaitingList;

	public String getIp() {
		return ip;
	}

	public void setIp(String ip) {
		this.ip = ip;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	@PostConstruct
	private void init() {
		if (syncDataWithVc()) {
			try {
				SystemConfig dictionary = systemConfigService.mergeByTypeName(
						new SystemConfig("syncSocket.ip", "localhost"), false);
				ip = dictionary.getValue();
				dictionary = systemConfigService.mergeByTypeName(new SystemConfig(
						"syncSocket.port", "5000"), false);
				port = new Integer(dictionary.getValue());
			} catch (Exception ex) {
				logger.error("Faield to get the config parameter", ex);
			}
			
			heartbeatScheduler = new TaskScheduler();
			heartbeatScheduler.setName("Send Sync Heartbeat");
			heartbeatScheduler.setCond("5000");			// 每5秒一次
			heartbeatScheduler.setActive(true);
			heartbeatScheduler.setCommand(this.getClass().getName() + ".sendHeartbeat");
			
			connectionScheduler = new TaskScheduler();
			connectionScheduler.setName("Init Sync Socket");
			connectionScheduler.setCond("60000");		// 每分钟一次
			connectionScheduler.setActive(true);
			connectionScheduler.setCommand(this.getClass().getName() + ".buildConnection");
			
			commandWaitingList = new ArrayList<>();		// TODO: use FIFO?
	
			systemConfigService.on(SystemConfig.TYPE_OTHER, "syncSocket.ip",
					new Executor<Boolean, String>() {
						@Override
						public Boolean execute(String newValue) {
							if (StringUtils.isNotEmpty(newValue)) {
								ip = newValue;
								disconnection();
							}
							return Boolean.TRUE;
						}
					});
			systemConfigService.on(SystemConfig.TYPE_OTHER, "syncSocket.port",
					new Executor<Boolean, String>() {
						@Override
						public Boolean execute(String newValue) {
							if (StringUtils.isNotEmpty(newValue)) {
								port = new Integer(newValue);
								disconnection();
							}
							return Boolean.TRUE;
						}
					});
			// Don't synchronize the local environment
			boolean isEclipseEnv = "true".equals(System.getProperty(SystemEnv.RUNNING_IN_ECLIPSE));
			if (!isEclipseEnv) {
				genericJPAService.executeAsync(new Executor<Boolean, String>() {
					@Override
					public Boolean execute(String p) {
						buildConnection();
						return true;
					}
				}, null);
			}
		}
	}
	
	private boolean syncDataWithVc() {
		return "true".equals(System.getProperty("raEntityInterceptor.syncDataWithVc"));
	}

	public Socket buildConnection() {
		if (client != null) {
			return client;
		}
		if ("localhost".equals(ip)) {			// Don't connect to local host
			return client;
		}
		
		// initialize the client if it's null
		try {
			synchronized (this) {
				if (client != null) {
					return client;
				}
				
				client = new Socket(ip, port);
				sendHeartbeat();
				
				if (!commandWaitingList.isEmpty()) {
					List<byte[]> list = new ArrayList<>(commandWaitingList);
					commandWaitingList.clear();
					for (byte[] command : list) {
						sendCommand(command);
					}
				}
				
				taskSchedulerService.runScheduler(heartbeatScheduler, "ra");
				taskSchedulerService.stopScheduler(connectionScheduler);
				
				clientThread = new Thread(new ReadHandlerThread(client));
				clientThread.start();
				if (logger.isDebugEnabled()) {
					logger.debug("Startup synchronization service successfully: " + ip + ":" + port);
				}
			}
		} catch (Exception e) {
			// only output the log, don't print the exception stack
			logger.error("Failed to startup the synchronization socket ("
					+ ip + ':' + port + "): " + e.getMessage());
			disconnection();
		}
		return client;
	}
	
	public synchronized void disconnection() {
		if (client == null) {
			taskSchedulerService.runScheduler(connectionScheduler, "ra");
			taskSchedulerService.stopScheduler(heartbeatScheduler);
			return ;
		}
		
		if (clientThread != null) {
			try {
				clientThread.interrupt();
			} catch (Exception ex) {
				logger.error("Stop thread", ex);
			}
		}
		client = null;
		clientThread = null;
		
		taskSchedulerService.runScheduler(connectionScheduler, "ra");
		taskSchedulerService.stopScheduler(heartbeatScheduler);
		logger.debug("Stop synchronization service");
	}

	/*
	 * 处理读操作的线程
	 */
	class ReadHandlerThread implements Runnable {
		private Socket client;

		public ReadHandlerThread(Socket client) {
			this.client = client;
		}

		@Override
		public void run() {
			InputStream dis = null;
			byte[] buffer = new byte[1];
			try {
				dis = client.getInputStream();
				while (true) {
					// 读取服务器端数据
					if (read(dis, buffer) == 0x21) {
						if (read(dis, buffer) == 0x03) {
							byte[] len = new byte[2];
							len[0] = read(dis, buffer);
							len[1] = read(dis, buffer);
							
							byte[] cmd = new byte[6];		// TODO: USE 'len'?
							for (int i = 0; i < cmd.length; i++) {
								cmd[i] = read(dis, buffer);
							}
							
							notifyCommand(cmd);
						}
					}
				}
			} catch (IOException e) {
				logger.error("Failed to receive the data, " + e.getMessage());
			} catch (ApplicationException e) {
				logger.error(e.getMessage());
			} finally {
				try {
					if (dis != null) {
						dis.close();
					}
				} catch (IOException e) {
					logger.error("Failed to close the input stream", e);
				}
			}
		}
		
		private byte read(InputStream dis, byte[] oneByte) throws IOException {
			if (dis.read(oneByte) == -1) {
				throw new ApplicationException("Server cloesed");
			} else {
				//System.err.print(new String(Hex.encodeHex(oneByte)) + ",");
				return oneByte[0];
			}
		}
	}

	public void onReceiveCommand(Executor<Boolean, Object[]> callback) {
		this.commandCallback = callback;
	}
	
	private void notifyCommand(byte[] command) {
		if (this.commandCallback != null) {
			ContextHolder contextHolder = ContextHolder.get();
			String oldSiteName = contextHolder.getSiteName();
			try {
				contextHolder.setSiteName(RaConstant.SITE_NAME);
				
				if (command != null && command.length == 6) {
					Object[] cmds = parseCommand(command);
		
					if (logger.isDebugEnabled()) {
						logger.debug("Receive command: " + toString(cmds));
					}
					this.commandCallback.execute(cmds);
				} else {
					if (logger.isWarnEnabled()) {
						logger.warn("Invalid command: " + command + ", Skip it");
					}
				}
			} finally {
				contextHolder.setSiteName(oldSiteName);
			}
		}
	}
	
	public void sendHeartbeat() {
		if (client != null) {
			// send the heart beat command
			boolean success = sendCommand(HEARTBEAT_COMMAND);
			if (success) {
				// TODO: receive the feedback?
			} else {
				errorCount++;
				
				if (errorCount >= 3) {
					disconnection();
					errorCount = 0;
				}
			}
		}
	}
	
	private synchronized boolean sendCommand(byte[] command) {
		boolean success = true;
		if (client != null) {
			try {
				OutputStream output = client.getOutputStream();
				output.write(command);
				output.flush();
			} catch (IOException e) {
				if (command == HEARTBEAT_COMMAND) {
					logger.error("Failed to send the heart beat" + "," + e.getMessage());
				} else {
					commandWaitingList.add(command);
					logger.error("Failed to send the data: " + toString(command) + "," + e.getMessage());
				}
				success = false;
			}
		} else {
			if (logger.isInfoEnabled()) {
				logger.info("Socket is closed, add command to waiting list");
			}
			commandWaitingList.add(command);
			if (commandWaitingList.size() > 10000) {		// max size is 10000
				commandWaitingList.clear();
			}
		}
		return success;
	}
	
	/**
	 * 
	 * @param cmds [A|U|D] + 2位表名对应的数字  + 剩余位表示的id数字， 共7位
	 * @return
	 */
	public boolean sendCommand(Collection<String> cmds) {
		if (logger.isDebugEnabled()) {
			logger.debug("Send command: " + cmds.toString());
		}
		byte[] command = buildCommand(cmds);
		return sendCommand(command);
	}
	
	private byte[] buildCommand(Collection<String> cmds) {
		int cmdLen = cmds.size() * 10;
		byte[] command = new byte[cmdLen];
		int begin = 0;
		for (String cmd : cmds) {
			command[begin++] = 0x20;
			command[begin++] = 0x03;
			command[begin++] = 0x06;
			command[begin++] = 0x00;
			command[begin++] = (byte)cmd.charAt(0);		// U/D
			// 小于100的整数, 转完后取4个字节的最低位
			command[begin++] = intToByte4(new Integer(cmd.substring(1,3)))[3];
			Integer id = new Integer(cmd.substring(3));
			// 4位id
			byte[] byteId = intToByte4(id);
			command[begin++] = byteId[0];
			command[begin++] = byteId[1];
			command[begin++] = byteId[2];
			command[begin++] = byteId[3];
		}
		return command;
	}
	
	private byte[] intToByte4(int i) {
		byte[] targets = new byte[4];
		targets[3] = (byte) (i & 0xFF);
		targets[2] = (byte) (i >> 8 & 0xFF);
		targets[1] = (byte) (i >> 16 & 0xFF);
		targets[0] = (byte) (i >> 24 & 0xFF);
		return targets;
	}

	private Object[] parseCommand(byte[] cmd) {
		byte c = cmd[0];
		int type = byte4ToInt(cmd, 1, 1);
		int id = byte4ToInt(cmd, 2, 4);
		return new Object[] {c, type, id};
	}

	private int byte4ToInt(byte[] bytes, int off, int len) {
		int ret = 0;
		if (len >= 4) {
			int b0 = bytes[off + len - 4] & 0xFF;
			ret = b0 << 24;
		}
		if (len >= 3) {
			int b1 = bytes[off + len - 3] & 0xFF;
			ret = ret | (b1 << 16);
		}
		if (len >= 2) {
			int b2 = bytes[off + len - 2] & 0xFF;
			ret = ret | (b2 << 8);
		}
		int b3 = bytes[off + len - 1] & 0xFF;
		return ret | b3;
	}
	
	private String toString(byte[] bytes) {
		StringBuilder sb = new StringBuilder(bytes.length);
		for (int i = 0; i < bytes.length; i++) {
			sb.append(bytes[i]).append(' ');
		}
		return sb.toString();
	}
	
	private String toString(Object[] objs) {
		StringBuilder sb = new StringBuilder(objs.length);
		for (int i = 0; i < objs.length; i++) {
			sb.append(objs[i]).append(' ');
		}
		return sb.toString();
	}
}